
#include <iostream>
#include <sstream>

#include <boost/thread/thread.hpp>
#include <boost/algorithm/string.hpp>

#include "bocat/common/ini.hpp"
#include "bocat/common/md5.hpp"
#include "bocat/common/base64.hpp"

#include "confab/packet.pb.h"

#include "broker.hpp"
#include "client.hpp"


namespace	confab	{
namespace	client	{


#define	MIN_SID	 1000
#define	MAX_SID  10000000

///////////////////////////////////////////////////////////////////////
//


Client::Client( const char* app_config )
{
	parseAppConfig( app_config );
	//parseLogConfig( log_config );

	initialize();

	_current_sid = MAX_SID;

	GOOGLE_PROTOBUF_VERIFY_VERSION;
}

Client::Client( const char* host, int port )
{
	char	port_[ 32 ];
	sprintf( port_, "%d", port );

	server_address_list.push_back( std::make_pair(host, port_) );

	//parseLogConfig( log_config );

	initialize();

	_current_sid = MAX_SID;

	GOOGLE_PROTOBUF_VERIFY_VERSION;
}


Client::~Client()
{
	finalize();

	google::protobuf::ShutdownProtobufLibrary();
}

///////////////////////////////////////////////////////////////////////
//

int   
Client::put( const std::string& key,  
		     const std::string& data )
{
	// make request packet
	confab::Packet   pkt_request;
	pkt_request.set_sid( this->getSid() );
	pkt_request.set_type( confab::Packet::REQUEST );
	pkt_request.set_op( confab::Packet::PUT );
	confab::Packet::Item*   item_p = NULL;
	item_p = pkt_request.add_item();
	item_p->set_key( key.c_str(), key.size() );
	item_p->set_value( data.c_str(), data.size() );

	bocat::common::MD5	m;
	m.update( key.c_str(), key.size() );
	const void*	p = static_cast< const void * >(m.digest());
	item_p->set_kpin( static_cast< const char* >(p), 16 );

	std::string pkt_str;
	if ( !pkt_request.SerializeToString( &pkt_str ) ) { 
		throw std::runtime_error("fail to serialize PUT packet to string.");
	}   
	std::string  request_b64 =  bocat::common::base64_encode(pkt_str) ;

	// get broker by table name
	Broker*	broker_p = getBroker();

	// call server via broker
	//      and get response packet
	std::string	response_b64;
	int	rc = this->callBroker( broker_p, request_b64, response_b64 );
	if (rc < 0) {
		std::ostringstream	msg;
		msg << "fail to call broker";
		throw std::runtime_error( msg.str() );
	}

	// parse response packet
	std::string	response = bocat::common::base64_decode< std::string >( response_b64 );

	confab::Packet   pkt_response;
	if ( !pkt_response.ParseFromString( response ) ) {
		std::ostringstream	msg;
		msg << "fail to parse from string (in base64) : " << response_b64;
		throw std::runtime_error( msg.str() );
	}

	if ( 1 != pkt_response.sid() ) {
		std::ostringstream	msg;
		msg << "ASSERT response packet (sid == 1), but got sid:" << pkt_response.sid();
		throw std::runtime_error( msg.str() );
	}
	if ( pkt_response.reqid() != pkt_request.sid() ) {
		std::ostringstream  msg;
		msg << "BAD reqid ( want:" << pkt_request.sid() << " got:" << pkt_response.reqid() << " )";
		throw std::runtime_error( msg.str() );
	}


	return	0;
}

int   
Client::get( const std::string& key,  
		           std::string& buf )
{
	// make request packet
	confab::Packet   pkt_request;
	pkt_request.set_sid( this->getSid() );
	pkt_request.set_type( confab::Packet::REQUEST );
	pkt_request.set_op( confab::Packet::GET );
	confab::Packet::Item*   item_p = NULL;
	item_p = pkt_request.add_item();
	item_p->set_key( key.c_str(), key.size() );

	bocat::common::MD5	m;
	m.update( key.c_str(), key.size() );
	const void*	p = static_cast< const void * >(m.digest());
	item_p->set_kpin( static_cast< const char* >(p), 16 );

	std::string pkt_str;
	if ( !pkt_request.SerializeToString( &pkt_str ) ) { 
		throw std::runtime_error("fail to serialize PUT packet to string.");
	}   
	std::string  request_b64 =  bocat::common::base64_encode(pkt_str) ;

	// get broker by table name
	Broker*	broker_p = getBroker();

	// call server via broker
	//      and get response packet
	std::string	response_b64;
	int	rc = this->callBroker( broker_p, request_b64, response_b64 );
	if (rc < 0) {
		std::ostringstream	msg;
		msg << "fail to call broker";
		throw std::runtime_error( msg.str() );
	}

	// parse response packet
	std::string	response = bocat::common::base64_decode< std::string >( response_b64 );

	confab::Packet   pkt_response;
	if ( !pkt_response.ParseFromString( response ) ) {
		std::ostringstream	msg;
		msg << "fail to parse from string (in base64) : " << response_b64;
		throw std::runtime_error( msg.str() );
	}

	if ( 1 != pkt_response.sid() ) {
		std::ostringstream	msg;
		msg << "ASSERT response packet (sid == 1), but got sid:" << pkt_response.sid();
		throw std::runtime_error( msg.str() );
	}
	if ( pkt_response.reqid() != pkt_request.sid() ) {
		std::ostringstream  msg;
		msg << "BAD reqid ( want:" << pkt_request.sid() << " got:" << pkt_response.reqid() << " )";
		throw std::runtime_error( msg.str() );
	}

	if ( pkt_response.item_size() < 1 ) {
		std::ostringstream	msg;
		msg << "BAD response : no items";
		throw std::runtime_error( msg.str() );
	}

	
	const confab::Packet::Item&	item = pkt_response.item( 0 );
	const std::string&	key_return = item.key();
	u_int32_t	status = item.status();
	if ( 200 == status ) {
		const std::string&	value = item.value();

        buf.assign( value.c_str(), value.size() );
		return value.size();
	}
	else if ( status >= 300 ) {
		return	-status;
	}

	return	0;
}

int   
Client::erase( const std::string& key )
{
	// make request packet
	confab::Packet   pkt_request;
	pkt_request.set_sid( this->getSid() );
	pkt_request.set_type( confab::Packet::REQUEST );
	pkt_request.set_op( confab::Packet::DELETE );
	confab::Packet::Item*   item_p = NULL;
	item_p = pkt_request.add_item();
	item_p->set_key( key.c_str(), key.size() );

	bocat::common::MD5	m;
	m.update( key.c_str(), key.size() );
	const void*	p = static_cast< const void * >(m.digest());
	item_p->set_kpin( static_cast< const char* >(p), 16 );

	std::string pkt_str;
	if ( !pkt_request.SerializeToString( &pkt_str ) ) { 
		throw std::runtime_error("fail to serialize PUT packet to string.");
	}   
	std::string  request_b64 =  bocat::common::base64_encode(pkt_str) ;

	// get broker by table name
	Broker*	broker_p = getBroker();

	// call server via broker
	//      and get response packet
	std::string	response_b64;
	int	rc = this->callBroker( broker_p, request_b64, response_b64 );
	if (rc < 0) {
		std::ostringstream	msg;
		msg << "fail to call broker";
		throw std::runtime_error( msg.str() );
	}

	// parse response packet
	std::string	response = bocat::common::base64_decode< std::string >( response_b64 );

	confab::Packet   pkt_response;
	if ( !pkt_response.ParseFromString( response ) ) {
		std::ostringstream	msg;
		msg << "fail to parse from string (in base64) : " << response_b64;
		throw std::runtime_error( msg.str() );
	}

	if ( 1 != pkt_response.sid() ) {
		std::ostringstream	msg;
		msg << "ASSERT response packet (sid == 1), but got sid:" << pkt_response.sid();
		throw std::runtime_error( msg.str() );
	}
	if ( pkt_response.reqid() != pkt_request.sid() ) {
		std::ostringstream  msg;
		msg << "BAD reqid ( want:" << pkt_request.sid() << " got:" << pkt_response.reqid() << " )";
		throw std::runtime_error( msg.str() );
	}

	return	0;
}


///////////////////////////////////////////////////////////////////////
//


int
Client::callBroker( Broker* broker_p, const std::string& request, std::string& response )
{
	int	retries = 10;
	int	retries_upbound = 86400;

	for ( int i=0; i<retries_upbound; i++ ) {
		try {
			int	rc = broker_p->call( request, response );
			return rc;
		}
		catch( std::exception& e) {
			if ( i < retries ) {
				boost::this_thread::sleep(boost::posix_time::milliseconds(1000));
				continue;
			}
			else {
				std::ostringstream	msg;
				msg << "call broker error (" << retries << " tries) -- " << e.what();
				throw std::runtime_error( msg.str() );
			}
		}
	}

	return	0;
}



Broker*
Client::getBroker( void )
{
	u_int32_t	sl_size = server_list.size();

	if ( sl_size < 1 ) {
		return NULL;
	}
	if ( sl_size == 1 ) {
		return server_list[ 0 ];
	}

    return  server_list[0];

    /*
	bocat::common::MD5	m;
	m.update( table, strlen(table) );
	const unsigned char*	p = m.digest();

	u_int32_t	x;
	x = (p[0] << 24) + (p[1] << 16) + (p[2] << 8) + p[3];

	u_int32_t	i = x % sl_size;
	Broker*	broker_p = server_list[ i ];

	return broker_p;
    */
}


int
Client::getSid()
{
	boost::mutex::scoped_lock   _scoped_lock( _current_sid_mutex );

	++_current_sid;

	if (_current_sid >= MAX_SID) {
		_current_sid = MIN_SID;
	}

	return	_current_sid;
}

///////////////////////////////////////////////////////////////////////
//

int
Client::initialize()
{
	for (size_t i=0; i<server_address_list.size(); i++) {
		server_address_type&	address = server_address_list[ i ];

		Broker*	p = new Broker( address.first, address.second );
		server_list.push_back( p );
	}

	return	0;
}

int
Client::finalize()
{
	for (size_t i=0; i<server_list.size(); i++) {
		Broker*	p = server_list[i];
		
		if (NULL != p) {
			delete p;
			server_list[i] = NULL;
		}
	}

	return	0;
}



///////////////////////////////////////////////////////////////////////
//

/*
 * Config Sample 
 *
 * [confab::client]
 * servers = server1:port1, server2:port2, server3:port3
 *
 * #### END ####
 *
 */
int		
Client::parseAppConfig( const char* filename )
{
	bocat::common::INIFile	ini( filename );

	std::string	SECTION = "confab::client";
	std::string	OPTION = "servers";
	
	std::string	value = ini[ SECTION ][ OPTION ];
	if (value.size() < 1) {
		std::ostringstream	msg;
		msg << "Empty value for [" << SECTION << "]::" << OPTION << " in " << filename;
		throw std::runtime_error( msg.str() );
	}

	std::vector< std::string >	fields;
	boost::split(fields, value, boost::is_any_of(","));

	for (size_t i=0; i<fields.size(); i++) {
		std::string&	field = fields[ i ];

		std::vector< std::string >	fs;
		boost::split(fs, field, boost::is_any_of(":"));
		if (fs.size() != 2) {
			// BAD field
			continue;
		}

		std::string	host = fs[0];
		std::string	port = fs[1];
		boost::trim( host );
		boost::trim( port );
		server_address_list.push_back( std::make_pair(host, port) );
	}

	return	0;
}

/*
 * Config Sample 
 *
 * [confab::client::logging]
 * log.error.enable = true
 * log.error.file.name = /path/to/file
 * log.error.file.rotate = 4
 * log.error.file.maxsize = 64MB
 *
 * log.info. ....
 *
 * log.debug. ....
 *
 * #### END ####
 *
 */
int		
Client::parseLogConfig( const char* filename )
{
	return	0;
}


	


}	// namespace client
}	// namespace confab


